package com.simple.mqtt.server.handler;

import com.simple.mqtt.server.config.MqttBrokerConfig;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.mqtt.MqttDecoder;
import io.netty.handler.codec.mqtt.MqttEncoder;
import io.netty.handler.timeout.IdleStateHandler;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * MQTT Broker Channel 初始化器
 *

 */
public class MqttBrokerChannelInitializer extends ChannelInitializer<SocketChannel> {
    private final MqttBrokerConfig config;
    private final ExecutorService businessExecutor;
    private final ScheduledExecutorService scheduledExecutor;

    public MqttBrokerChannelInitializer(MqttBrokerConfig config, ExecutorService businessExecutor, ScheduledExecutorService scheduledExecutor) {
        this.config = config;
        this.businessExecutor = businessExecutor;
        this.scheduledExecutor = scheduledExecutor;
    }

    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();

        pipeline.addLast("decoder", new MqttDecoder(config.getMaxBytesInMessage()));
        pipeline.addLast("encoder", MqttEncoder.INSTANCE);

        long keepAliveSeconds = config.getKeepAliveTimeout() / 1000;
        pipeline.addLast("idleStateHandler",
                new IdleStateHandler(0, 0, (int) (keepAliveSeconds * 1.5), TimeUnit.SECONDS));

        pipeline.addLast("handler", new MqttBrokerHandler(config, businessExecutor, scheduledExecutor));
    }
}

